package cn.gwm.flink.streaming.function.process;

import cn.gwm.flink.streaming.beans.estimateSoh.EstimateSohRecordPhev;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONUtil;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Date;

/**
 * soh估算输出处理函数
 * @author GW00283474
 */
public class EstimateSohOutPhevProcessFunction extends KeyedProcessFunction<String, EstimateSohRecordPhev,String> {

    private OutputTag outputTag;

    public EstimateSohOutPhevProcessFunction(OutputTag outputTag) {
        this.outputTag = outputTag;
    }

    @Override
    public void processElement(EstimateSohRecordPhev record, KeyedProcessFunction<String, EstimateSohRecordPhev, String>.Context ctx, Collector<String> out) throws Exception {
        //填充处理结束时间
        Date date=new Date();
        record.setHandleEndDate(DateUtil.format(date, DatePattern.NORM_DATE_PATTERN));
        record.setHandleEndTime(DateUtil.format(date,DatePattern.NORM_DATETIME_PATTERN));
        //发送到kafka
        out.collect(JSONUtil.toJsonStr(record));
        //发送到hive侧输出流
        ctx.output(outputTag,record.getPropertyInfo("\t"));
    }
}
